/*
* Copyright (C) 2010-2012 Magnus Skjegstad
* 
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
* 
* The latest version of this library can be downloaded from http://mist-pubsub.googlecode.com
* 
*/
package com.skjegstad.mist2.datatypes;

import com.skjegstad.mist2.hash.MurmurHash;
import java.util.Arrays;

/**
 *
 * @author Magnus Skjegstad
 */
public class DataElement {
    protected NodeIdentifier sourceNode;
    protected byte[] payload;
    private long age;
    protected long expireTime;
    private long receiveTime;
    protected TopicIdentifier topic;
    protected long seqNr;
    private Long memoizedHash = null;
    protected boolean invalid = false;

    public DataElement(NodeIdentifier sourceNode, long seqNr, byte[] data, TopicIdentifier topic, long age, long expireTime) {
        this.sourceNode = sourceNode;
        this.payload = data;
        this.age = age;
        this.expireTime = expireTime;
        this.seqNr = seqNr;
        this.topic = topic;
        this.invalid = false;
        receiveTime = System.currentTimeMillis();
    }

    public TopicIdentifier getTopic() {
        return topic;
    }

    public byte[] getPayload() {
        return this.payload;
    }

    public int getPayloadLen() {
        return this.payload.length;
    }

    public long getAge() {
        return (System.currentTimeMillis() - receiveTime) + age;
    }

    public long getExpireTime() {
        return expireTime;
    }

    public long getReceiveTime() {
        return receiveTime;
    }

    public NodeIdentifier getSourceNode() {
        return sourceNode;
    }

    public boolean isExpired() {
        return getExpireTime() > 0 && getAge() > getExpireTime();
    }

    public void setExpireTime(long maximumAgeInMilliseconds) {
        this.expireTime = maximumAgeInMilliseconds;
    }        

    @Override
    public boolean equals(Object obj) {
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        final DataElement other = (DataElement) obj;
        if (this.sourceNode != other.sourceNode && (this.sourceNode == null || !this.sourceNode.equals(other.sourceNode))) {
            return false;
        }
        if (!Arrays.equals(this.payload, other.payload)) {
            return false;
        }
        if (this.topic != other.topic && (this.topic == null || !this.topic.equals(other.topic))) {
            return false;
        }
        if (this.seqNr != other.seqNr) {
            return false;
        }
        return true;
    }

    @Override
    public int hashCode() {
        int hash = 7;
        hash = 23 * hash + (this.sourceNode != null ? this.sourceNode.hashCode() : 0);
        hash = 23 * hash + Arrays.hashCode(this.payload);
        hash = 23 * hash + (this.topic != null ? this.topic.hashCode() : 0);
        hash = 23 * hash + (int) (this.seqNr ^ (this.seqNr >>> 32));
        return hash;
    }

    public long getHash() {
        if (memoizedHash == null)
            memoizedHash = getHash(sourceNode, seqNr, 0, this.payload, 0, this.payload.length);
        return memoizedHash;
    }

    public long getSeqNr() {
        return seqNr;
    }

    public static long getHash(NodeIdentifier sourceNode, long seqNr, int fragmentNr, byte[] payload, int start, int len) {
        long seed = 41 * sourceNode.getNodeIdentifier();
        seed = seed + 41 * seqNr;
        seed = seed + 41 * fragmentNr;
        long hash = MurmurHash.hash64(payload, start, len, seed);
        //System.err.println("hashing from " + start + ", len " + len + ", seed " +seed + ", res:" + hash);
        return hash;
    }

    public boolean isInvalid() {
        return invalid;
    }

    public void invalidate() {
        this.invalid = false;
    }

    
}
